package studio.raptor.databus.client;

import com.alibaba.fastjson.JSON;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Pattern;
import javax.jms.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.ActiveMQConnectionFactory;
import studio.raptor.databus.client.config.BrokerGroup;
import studio.raptor.databus.command.*;

import javax.jms.*;
import java.util.List;

/**
 * 队列消息生产者，每个生产者只能发送消息到一个指定的主题
 */
public final class DatabusProducer {

  private static Logger log = LoggerFactory.getLogger(DatabusProducer.class);
  private DatabusConfig databusConfig;
  private DatabusProducerBuilder databusProducerBuilder;

  private HashMap<Integer/*brokerGroupId*/, Connection> brokerGroupIdConMap =
      new HashMap<Integer/*brokerGroupId*/, Connection>(5);
  private HashMap<Integer/*brokerGroupId*/, Session> brokerGroupIdSessMap =
      new HashMap<Integer/*brokerGroupId*/, Session>(5);
  private HashMap<Integer/*brokerGroupId*/, MessageProducer> brokerGroupIdProMap =
      new HashMap<Integer/*brokerGroupId*/, MessageProducer>(5);

  /**
   * 在使用sendWithNoCommit方法时正在发送中的事件类型
   */
  private String currentSendingEventNameKindId = null;

  private static final String DELIVERTIME_FORMAT_REGEXP = "(19[0-9]{2})|(2[0-9]{3})-(0[1-9]|1[012])-(0[1-9]|[12][0-9]|3[01]) ([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]";

  private static final Pattern PATTERN = Pattern.compile(DELIVERTIME_FORMAT_REGEXP);
  //private HashMap<Integer/*eventKindId*/, Integer/*brokerGroupId*/> eventKindIdGrpIdMap = new HashMap<Integer/*eventKindId*/, Integer/*brokerGroupId*/>();

  private List<BrokerGroup> brokerGroups = new ArrayList<BrokerGroup>();

  /**
   * 构造方法,对包内其他类开放
   */
  DatabusProducer(DatabusConfig databusConfig) {
    this.databusConfig = databusConfig;
  }

  /**
   * 通过DatabusConfig实例和DatabusProducerBuilder实例生成消息发送者
   *
   * @param databusConfig databus配置实例
   * @param producerBuilder producerBuilder配置实例
   * @param brokerGroups brokergroup配置信息
   */
  DatabusProducer(DatabusConfig databusConfig, DatabusProducerBuilder producerBuilder,
      List<BrokerGroup> brokerGroups) {
    this.databusConfig = databusConfig;
    this.databusProducerBuilder = producerBuilder;
    //this.eventKindIdGrpIdMap = eventKindIdGrpIdMap;
    this.brokerGroups = brokerGroups;
  }

  /**
   * 批量发送消息事件.<br>
   * 1、事件集合中事件字段违反约束时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 2、事件集合中事件未允许发送授权时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 3、事件集合中必须保证所有的事件eventKindId相同。
   *
   * @param events 事件列表
   */
  public void send(List<Event> events) throws JMSException {
    if (events == null) {
      throw new JMSException("事件不可为空");
    }
    if (events.size() == 0) {
      return;
    }
    if( !checkEventsSize(events) ){
      throw new JMSException("您发送的所有事件消息内容过大，超过了每次允许发送最大值【"+this.databusProducerBuilder.getEventMaxSize()+"KB】");
    }
    Integer eventKindIdTemp = events.get(0).getEventBasicData().getKindId();
    String eventNameTemp = events.get(0).getEventBasicData().getName();

    //检查当前集合是否存在不同eventKindId的事件
    for (int i = 0; i < events.size(); i++) {
      if (events.get(i).getEventBasicData().getKindId().intValue() != eventKindIdTemp.intValue()) {
        throw new JMSException("批量发送事件不可发送不同种类的事件！");
      }
      //是否有权限发送事件
      if (!checkSendAuthorization(events.get(i))) {
        log.info(
            "您无权发送事件:[" + events.get(i).getEventBasicData().getName() + "-" + events.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
        throw new JMSException(
            "您无权发送事件:[" + events.get(i).getEventBasicData().getName() + "-" + events.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
      }
      String validResult = events.get(i).validate();
      if (!Objects.equals("", validResult)) {
        throw new JMSException(validResult);
      }
    }
    //相同类型的事件，使用相同的生产者发送消息
    Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub()
        .get(eventNameTemp + "_" + eventKindIdTemp);
    if (!brokerGroupIdProMap.containsKey(brokerGroupId) || !brokerGroupIdSessMap
        .containsKey(brokerGroupId)) {
      //若不存在，需要对其进行初始化
      initProducer(brokerGroupId, brokerGroupIdProMap,
          brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);
    }

    for (int i = 0; i < events.size(); i++) {
      Message msg = this.transFromEvent(events.get(i),brokerGroupId);
      String eventName = events.get(i).getEventBasicData().getName();
      Integer eventKindId = events.get(i).getEventBasicData().getKindId();
      String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
      String virtualTopic =
          DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId
              .intValue();

      Destination destination = this.brokerGroupIdSessMap.get(brokerGroupId)
          .createTopic(virtualTopic);
      this.brokerGroupIdProMap.get(brokerGroupId).send(destination, msg);
    }

    this.brokerGroupIdSessMap.get(brokerGroupId).commit();
  }

  /**
   * 批量发送消息事件,且每个事件都将会携带属性<br>
   * 1、事件集合中事件字段违反约束时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 2、事件集合中事件未允许发送授权时，全部事件发送失败。建议在批量发送事件之前自行校验<br>
   * 3、事件集合中必须保证所有的事件eventKindId相同。
   *
   * @param eventProperties
   * @throws JMSException
   */
  public void send(ArrayList<EventProperty> eventProperties) throws JMSException {
    if (eventProperties == null) {
      throw new JMSException("事件不可为空");
    }
    if (eventProperties.size() == 0) {
      return;
    }
    List<Event> events = new ArrayList<>();
    for ( EventProperty eventProperty :eventProperties ){
      events.add(eventProperty.getEvent());
    }
    if( !checkEventsSize(events) ){
      throw new JMSException("您发送的所有事件消息内容过大，超过了每次允许发送最大值【"+this.databusProducerBuilder.getEventMaxSize()+"KB】");
    }
    Integer eventKindIdTemp = eventProperties.get(0).getEvent().getEventBasicData().getKindId();
    String eventNameTemp = eventProperties.get(0).getEvent().getEventBasicData().getName();

    //检查当前集合是否存在不同eventKindId的事件
    for (int i = 0; i < events.size(); i++) {
      if (events.get(i).getEventBasicData().getKindId().intValue() != eventKindIdTemp.intValue()) {
        throw new JMSException("批量发送事件不可发送不同种类的事件！");
      }
      //是否有权限发送事件
      if (!checkSendAuthorization(events.get(i))) {
        log.info(
            "您无权发送事件:[" + events.get(i).getEventBasicData().getName() + "-" + events.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
        throw new JMSException(
            "您无权发送事件:[" + events.get(i).getEventBasicData().getName() + "-" + events.get(i)
                .getEventBasicData().getKindId() + "],若需发送请先申请");
      }
      String validResult = events.get(i).validate();
      if (!Objects.equals("", validResult)) {
        throw new JMSException(validResult);
      }
    }
    //相同类型的事件，使用相同的生产者发送消息
    Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub()
        .get(eventNameTemp + "_" + eventKindIdTemp);
    if (!brokerGroupIdProMap.containsKey(brokerGroupId) || !brokerGroupIdSessMap
        .containsKey(brokerGroupId)) {
      //若不存在，需要对其进行初始化
      initProducer(brokerGroupId, brokerGroupIdProMap,
          brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);
    }

    for (int i = 0; i < eventProperties.size(); i++) {
      Event event = eventProperties.get(i).getEvent();
      MessageProperty messageProperty = eventProperties.get(i).getMessageProperty();
      Message msg = this.transFromEvent(event,brokerGroupId);
      //为待发送的事件设置消息属性
      this.setProperty4Message(msg,messageProperty);
      String eventName = event.getEventBasicData().getName();
      Integer eventKindId = event.getEventBasicData().getKindId();
      String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
      String virtualTopic =
          DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId
              .intValue();

      Destination destination = this.brokerGroupIdSessMap.get(brokerGroupId)
          .createTopic(virtualTopic);
      this.brokerGroupIdProMap.get(brokerGroupId).send(destination, msg);
    }

    this.brokerGroupIdSessMap.get(brokerGroupId).commit();
  }

  /**
   * 发送携带消息属性的事件
   * @param event 事件实例
   * @param messageProperty 事件属性实例
   * @throws JMSException
   */
  public void send(Event event,MessageProperty messageProperty) throws JMSException{
    String validResult = event.validate();

    if (!Objects.equals("", validResult)) {
      throw new JMSException(validResult);
    }
    if( !checkEventSize(event) ){
      throw new JMSException("您发送的事件消息内容过大，超过了每次允许发送最大值【"+this.databusProducerBuilder.getEventMaxSize()+"KB】");
    }

    Integer eventKindId = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();

    //获取当前事件对应的brokerGroupId
    Integer brokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub()
        .get(eventName + "_" + eventKindId);

    //初始化发送当前事件的生产者
    initProducer(brokerGroupId, brokerGroupIdProMap,
        brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);
    //将待发送事件转换为消息对象
    Message msg = this.transFromEvent(event,brokerGroupId);
    //为待发送的消息设置属性，供消费者过滤
    this.setProperty4Message(msg,messageProperty);
    Destination destination =
        createDestination(this.brokerGroupIdSessMap.get(brokerGroupId), eventName, eventKindId);

    reallySend(this.brokerGroupIdProMap.get(brokerGroupId), destination, msg);
    reallyCommit(this.brokerGroupIdSessMap.get(brokerGroupId));
  }

  /**
   * 发送消息事件，消息事件不携带属性
   *
   * @param event 事件
   */
  public void send(Event event) throws JMSException {
    this.send(event,null);
  }

  /**
   * 真正用于发送消息的方法
   *
   * @param producer 消息生产者
   * @param destination 消息目的地
   * @param message 消息载体对象
   */
  private void reallySend(MessageProducer producer, Destination destination, Message message)
      throws JMSException {
    producer.send(destination, message);
  }

  /**
   * 会话提交方法
   *
   * @param session 会话
   */
  private void reallyCommit(Session session) throws JMSException {
    session.commit();
  }

  /**
   * 发送事件但不进行提交(提交之前所发送的事件必须为相同类型)
   * <p>与方法<code>commit()</code>或<code>rollback</code>配合使用</p>
   * <p>此方法在执行commit或者rollback之前,事件种类ID必须一致，否则强制回滚</p>
   *
   * @param event 待发送的事件
   */
  public void sendWithNoCommit(Event event) throws JMSException {
    this.sendWithNoCommit(event,null);
  }

  /**
   * 发送事件但不进行提交(提交之前所发送的事件必须为相同类型)
   * <p>与方法<code>commit()</code>或<code>rollback</code>配合使用</p>
   * <p>此方法在执行commit或者rollback之前,事件种类ID必须一致，否则强制回滚</p>
   *
   * @param event 待发送事件
   * @param messageProperty 待发送事件的属性，若此值为空，则不会在发送的消息上设置属性
   * @throws JMSException
   */
  public void sendWithNoCommit(Event event,MessageProperty messageProperty) throws JMSException {
    String validResult = event.validate();
    if (!Objects.equals("", validResult)) {
      throw new JMSException(validResult);
    }

    if( !checkEventSize(event) ){
      throw new JMSException("您发送的事件消息内容过大，超过了每次允许发送最大值【"+this.databusProducerBuilder.getEventMaxSize()+"KB】");
    }

    Integer eventKindId = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();

    //获取当前事件对应的brokerGroupId
    Integer brokerGroupId =
        this.databusConfig.getBizPubSubEventInfo().getPub().get(eventName + "_" + eventKindId);

    //校验发送的事件是否有不同类型
    if (Objects.equals(null, this.currentSendingEventNameKindId)) {
      //当前发送的消息类型为空时赋值，表示开始发送当前类型事件
      this.currentSendingEventNameKindId = eventName + "_" + eventKindId;
    } else {
      //若发现发送的事件不一致，回滚
      if (!this.currentSendingEventNameKindId.equals(eventName + "_" + eventKindId)) {
        Integer preBrokerGroupId = this.databusConfig.getBizPubSubEventInfo().getPub().get(
            currentSendingEventNameKindId);
        this.brokerGroupIdSessMap.get(preBrokerGroupId).rollback();
        this.currentSendingEventNameKindId = null;
        throw new JMSException("使用事务性生产者发送消息，同一次事务中只能发送同一种消息类型");
      }
    }

    initProducer(brokerGroupId, brokerGroupIdProMap,
        brokerGroupIdSessMap, brokerGroupIdConMap, databusProducerBuilder);

    Message msg = this.transFromEvent(event,brokerGroupId);

    //设置消息的属性
    this.setProperty4Message(msg,messageProperty);

    Destination destination =
        createDestination(this.brokerGroupIdSessMap.get(brokerGroupId), eventName, eventKindId);

    reallySend(this.brokerGroupIdProMap.get(brokerGroupId), destination, msg);

  }
  /**
   * 将已经发送的事件进行提交操作
   */
  public void commit() throws JMSException {
    if (this.currentSendingEventNameKindId != null) {
      Integer currentSendBrokerGroupId =
          this.databusConfig.getBizPubSubEventInfo().getPub().get(currentSendingEventNameKindId);
      reallyCommit(this.brokerGroupIdSessMap.get(currentSendBrokerGroupId));
      //重置当前生产者可以发送任何消息种类
      this.currentSendingEventNameKindId = null;
    }
  }


  /**
   * 生产者回滚
   */
  public void rollback() throws JMSException {

    if (this.currentSendingEventNameKindId != null) {
      Integer currentSendBrokerGroupId =
          this.databusConfig.getBizPubSubEventInfo().getPub().get(currentSendingEventNameKindId);
      this.brokerGroupIdSessMap.get(currentSendBrokerGroupId).rollback();
      //重置当前生产者可以发送任何消息种类
      this.currentSendingEventNameKindId = null;
    }

  }

  /**
   * 校验当前生产者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   *
   * @param event 事件对象实例
   * @return 是否授权
   * @deprecated As of databus-client version 0.1.3-SNAPSHOT, replaced by
   * <code>checkSendAuthorization(EventProtoBuf.Event event)</code>.
   */
  @Deprecated
  public boolean sendAuthorizationCheck(EventProtoBuf.Event event) {
    Integer eventSeq = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();
    return sendAuthorizationCheck(eventSeq, eventName);
  }


  /**
   * 校验当前生产者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   *
   * @param eventKindId 事件种类Id
   * @param eventName 事件名称
   * @deprecated As of databus-client version 0.1.3-SNAPSHOT, replaced by
   * <code>checkSendAuthorization(Integer eventKindId,String eventName)</code>.
   */
  @Deprecated
  public boolean sendAuthorizationCheck(Integer eventKindId, String eventName) {

    return this.databusConfig.getBizPubSubEventInfo().getPub()
        .containsKey(eventName + "_" + eventKindId);
  }

  /**
   * 校验消费者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   *
   * @param eventKindId 事件类型ID
   * @param eventName 事件名称
   * @return 是否有授权
   */
  public boolean checkSendAuthorization(Integer eventKindId, String eventName) {
    return this.databusConfig.getBizPubSubEventInfo().getPub()
        .containsKey(eventName + "_" + eventKindId);
  }

  /**
   * 校验消费者是否有权发送指定事件，当且仅当事件序号和事件名称都满足时校验通过
   *
   * @param event 事件对象实例
   * @return 是否有授权
   */
  public boolean checkSendAuthorization(Event event) {
    Integer eventSeq = event.getEventBasicData().getKindId();
    String eventName = event.getEventBasicData().getName();
    return checkSendAuthorization(eventSeq, eventName);
  }

  /**
   * 关闭所有生产者及其对应的会话和连接
   */
  public void close() {
    try {
      //关闭所有的生产者
      Set<Entry<Integer, MessageProducer>> producers = this.brokerGroupIdProMap.entrySet();
      Iterator<Entry<Integer, MessageProducer>> producerIt = producers.iterator();
      while (producerIt.hasNext()) {
        MessageProducer producer = producerIt.next().getValue();
        if (producer != null) {
          producer.close();
        }
      }
      //关闭所有的会话
      Set<Entry<Integer, Session>> sessions = this.brokerGroupIdSessMap.entrySet();
      Iterator<Entry<Integer, Session>> sessionIt = sessions.iterator();
      while (sessionIt.hasNext()) {
        Session session = sessionIt.next().getValue();
        if (session != null) {
          session.close();
        }
      }
      // 关闭所有的连接
      Set<Entry<Integer, Connection>> connections = this.brokerGroupIdConMap.entrySet();
      Iterator<Entry<Integer, Connection>> connectionIt = connections.iterator();
      while (connectionIt.hasNext()) {
        Connection connection = connectionIt.next().getValue();
        if (connection != null) {
          connection.close();
        }
      }
    } catch (JMSException e) {
      log.error(e.getMessage(), e);
    }
  }


  /**
   * 创建指定发送消息到指定Broker的消费者,若已创建则忽略
   *
   * @param brokerGroupId broker组ID
   * @param brokerGroupIdProMap 存放brokerGroupId和生产者的map
   * @param brokerGroupIdSessMap 存放brokerGroupId和会话的map
   * @param brokerGroupIdConMap 存放brokerGroupId和连接的map
   * @param producerBuilder 包含了生产者配置的实例
   */
  private void initProducer(Integer brokerGroupId,
      HashMap<Integer/*brokerGroupId*/, MessageProducer> brokerGroupIdProMap,
      HashMap<Integer/*brokerGroupId*/, Session> brokerGroupIdSessMap,
      HashMap<Integer/*brokerGroupId*/, Connection> brokerGroupIdConMap,
      DatabusProducerBuilder producerBuilder) throws JMSException {

    //若指定的生产者及其对应的会话和连接已经存在，直接返回,不再进行初始化
    if (brokerGroupIdProMap.containsKey(brokerGroupId) && brokerGroupIdSessMap
        .containsKey(brokerGroupId)) {
      return;
    }

    String userName = "";
    String password = "";
    String brokerURL = DatabusConfig.LOAD_CONFIG_FAILURE;
    //依据brokerGroupId选择配置的url
    for (BrokerGroup brokerGroup : this.brokerGroups) {
      if (brokerGroupId.intValue() == brokerGroup.getBrokerGroupId()) {
        userName = brokerGroup.getUserName();
        password = brokerGroup.getPassword();
        brokerURL = brokerGroup.getBrokerURL();
        break;
      }
    }
    if (DatabusConfig.LOAD_CONFIG_FAILURE.equals(brokerURL)) {
      throw new JMSException("当前事件消息未配置发送目的地或配置了错误的目的地");
    }
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    connectionFactory.setUserName(userName);
    connectionFactory.setPassword(password);
    String customizedBrokerURL = producerBuilder.transform2URLConfig();
    connectionFactory.setBrokerURL(brokerURL + "?" + customizedBrokerURL);

    // 配置生产者客户端中的连接、会话
    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection
        .createSession(producerBuilder.isTransactedSession(), producerBuilder.getAcknowledgeMode());
    MessageProducer producer = session.createProducer(null);
    producer.setTimeToLive(producerBuilder.getMsgTimeToLive() * 24 * 3600 * 1000);
    brokerGroupIdConMap.put(brokerGroupId, connection);
    brokerGroupIdSessMap.put(brokerGroupId, session);
    brokerGroupIdProMap.put(brokerGroupId, producer);

  }

  /**
   * 创建 Destination
   *
   * @param session 会话
   * @param eventName 事件名称
   * @param eventKindId 事件种类ID
   * @return 消息目的地destination
   */
  private Destination createDestination(Session session, String eventName, Integer eventKindId)
      throws JMSException {
    String topicSalt = TopicSalt.createSalt(eventName, eventKindId);
    String virtualTopic =
        DatabusConfig.VIRTUAL_TOPIC_PRE + eventName + "_" + topicSalt + "_" + eventKindId
            .intValue();
    Destination destination = session.createTopic(virtualTopic);
    return destination;
  }


  /**
   * 将Event对象转换为可以直接发送的消息接口
   * @param event
   * @return
   */
  /**
   * 发送消息时，用于将Event对象转换为可以直接发送的消息
   * @param event 事件对象
   * @param brokerGroupId 当前发送的消息要发送到的broker
   * @return
   * @throws JMSException
   */
  private Message transFromEvent(Event event,Integer brokerGroupId) throws JMSException {
    Message msg = null;
    switch (this.databusConfig.TRANSPORTTYPE){

      case EVENT_TRANSPORT_TYPE_ORIGINAL:
        //原始的传输对象方式
        msg = this.brokerGroupIdSessMap.get(brokerGroupId).createObjectMessage(event);
        break;
      case EVENT_TRANSPORT_TYPE_TEXT:
        //使用序列化后的文本格式传输
        msg = this.brokerGroupIdSessMap.get(brokerGroupId).createTextMessage(JSON.toJSONString(event));
        break;
      case EVENT_TRANSPORT_TYPE_SERIALIZATION:
        EventProtoBuf.Event eventBuf = EventMessage.transFromEvent(event);
        //使用序列化对象传输
        BytesMessage bytesMessage = this.brokerGroupIdSessMap.get(brokerGroupId).createBytesMessage();
        bytesMessage.writeBytes(eventBuf.toByteArray());
        msg = bytesMessage;
        break;
      default:
        break;
    }

    return msg;
  }

  /**
   * 校验事件大小
   * @param event
   * @param eventSize 事件大小，单位为KB
   * @return
   */
  private boolean checkEventSize(Event event,int eventSize){
    String jsonStr = JSON.toJSONString(event);
    if( jsonStr.getBytes().length > eventSize*1024 ){
      return false;
    }
    return true;
  }

  /**
   * 校验事件大小
   * @param event
   * @return 若事件大小大于配置的最大值，返回false；若事件大小于配置的大小值，返回true，表示可以发送
   */
  public boolean checkEventSize(Event event){
    int eventSize = this.databusProducerBuilder.getEventMaxSize();
    return checkEventSize(event,eventSize);
  }


  /**
   *
   * @param events 事件集合
   * @param eventSize 事件集合的最大值，单位为KB
   * @return
   */
  private boolean checkEventsSize(List<Event> events,int eventSize){
    int totalSize = 0;
    for( Event event:events ){
      String jsonStr = JSON.toJSONString(event);
      totalSize += jsonStr.getBytes().length;
      if( totalSize > eventSize*1024 ){
        return false;
      }
    }
    return true;
  }

  /**
   * 校验待发送的集合大小是否超过限制
   * @param events
   * @return 若事件集合中所有消内容大于配置的允许消息发送消息最大值，返回false，表示不可发送,返回true则表示允许发送
   */
  public boolean checkEventsSize(List<Event> events){
    int eventSize = this.databusProducerBuilder.getEventMaxSize();
    return checkEventsSize(events,eventSize);
  }

  /**
   * 为待发送的消息设置消息属性
   * @param message 待发送的消息
   * @param messageProperty 消息属性，若此值为空，则不对消息设置属性
   */
  private void setProperty4Message(Message message,MessageProperty messageProperty)
      throws JMSException {
    if( Objects.equals(null,messageProperty) ){
      return ;
    }
    //设置String类型属性
    HashMap<String,String> stringP = messageProperty.getStringProperty();
    if(!Objects.equals(null,stringP) ){
      Set<Entry<String,String>> stringSet = stringP.entrySet();
      for (Entry<String,String> entry:stringSet){
        message.setStringProperty(entry.getKey(),entry.getValue());
      }
    }

    //设置int类型属性
    HashMap<String,Integer> intP = messageProperty.getIntProperty();
    if(!Objects.equals(null,intP) ){
      Set<Entry<String,Integer>> intSet = intP.entrySet();
      for (Entry<String,Integer> entry:intSet){
        message.setIntProperty(entry.getKey(),entry.getValue());
      }
    }
    //设置long类型属性
    HashMap<String,Long> longP = messageProperty.getLongProperty();
    if(!Objects.equals(null,longP) ){
      Set<Entry<String,Long>> intSet = longP.entrySet();
      for (Entry<String,Long> entry:intSet){
        message.setLongProperty(entry.getKey(),entry.getValue());
      }
    }

  }
}
